package cqut.wys.client;

import cqut.wys.config.KafkaProducerConfig;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

/**
 * --------神兽出没--------/
 *   ┌─┐     ┌─┐
 * ┌─┘─┴─────┘─┴─┐
 * │      ─      │
 * │  ┬─┘   └─┬  │
 * │             │
 * │      ┴      │  Code is far away from bug with the animal protecting
 * └───┐      ┌──┘  神兽保佑,代码无bug
 *     │      │
 *     │      └──────┐
 *     │             ├┐
 *     │             ┌┘
 *     └┐ ┐ ┌───┬─┐ ┌┘
 *      │ ┤ ┤   │ ┤ ┤
 *      └─┴─┘   └─┴─┘
 * --------感觉萌萌哒--------/
 *
 * @author wuyoushan
 * @date 2018/7/8
 */
public class StringClient {

    private static final String BROKER_LIST = "localhost:9092";

    private static final String TOPIC = "test";

    public static void main(String[] args) {
        Producer producer = new KafkaProducerConfig()
                .setKeySerializer("org.apache.kafka.common.serialization.StringSerializer")
                .setValueSerializer("org.apache.kafka.common.serialization.StringSerializer")
                .setClientId("hidden-producer-client-id-1")
                .setBrokerList(BROKER_LIST)
                .buildProducer();

        while (true) {
            String message = "kafka_message-" + System.currentTimeMillis() + "";
            ProducerRecord<String, String> producerRecord = new ProducerRecord<>(TOPIC, message);
            try {
                Future<RecordMetadata> future = producer.send(producerRecord, new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata metadata, Exception exception) {
                        System.out.println(metadata.offset() + "    ");
                        System.out.println(metadata.topic() + "    ");
                        System.out.println(metadata.partition());
                    }
                });
            } catch (Exception e) {
                e.printStackTrace();
            }
            try {
                TimeUnit.MILLISECONDS.sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
